package com.galeno.udfs;


import com.galeno.pojo.EventStateBean;
import com.galeno.pojo.LogBean;
import com.galeno.pojo.MatchResult;
import com.galeno.pojo.RuleCondition;
import com.galeno.service.QueryService;
import com.galeno.utils.RuleComparator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/11/1121:20
 */
public class RuleMatchFunction extends KeyedProcessFunction<String, LogBean, MatchResult> {
    private QueryService queryService;
    private RuleCondition ruleCondition;
    transient   ListState<EventStateBean> eventState;
    public RuleMatchFunction(){

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * 获取客户端发送的全局参数 通过运行时上下文,获取客户端发送的全局参数
         */
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //初始化或者恢复状态
        ListStateDescriptor<EventStateBean> eventStateDescriptor = new ListStateDescriptor<>("event-state", EventStateBean.class);
        //设置TTL
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).build();
        eventStateDescriptor.enableTimeToLive(ttlConfig);
        //恢复状态
         eventState = getRuntimeContext().getListState(eventStateDescriptor);


        queryService=new QueryService();
        //将参数传入init方法中
        queryService.init(parameterTool,eventState);

        //获取事先定义好的规则,以后可以从广播状态中获取
        ruleCondition= RuleComparator.getRule();



    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {
       //时间数据添加到状态
        eventState.add(bean.toEventStateBean());



        //匹配规则
        boolean match = queryService.isMatch(bean, ruleCondition);
        //匹配出结结果
        if (match){
            //输出结果
            out.collect(new MatchResult(bean.getDeviceId(),bean.getTimeStamp(),System.currentTimeMillis(),ruleCondition.getRuleId()));
        }



    }


    @Override
    public void close() throws Exception {
        queryService.destroy();
    }
}
